package net.bwie.jtp.dws.log.Function;

import net.bwie.jtp.dws.log.bean.PageViewBean;
import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


/**
 * 对窗口中的数据进行输出操作处理，直接获取增加计算中间结果即可
 */
public class PageViewReportWindowFunction implements WindowFunction<PageViewBean, String, String, TimeWindow> {


    @Override
    public void apply(String s, TimeWindow window, Iterable<PageViewBean> input, Collector<String> out) throws Exception {

        //1-获取窗口开始时间和结束时间
        String windowStartTime = DateTimeUtil.convertLongToString(window.getStart(), "yyyy-MM-dd HH:mm:ss");
        String windowEndTime = DateTimeUtil.convertLongToString(window.getEnd(), "yyyy-MM-dd HH:mm:ss" );
// 2获取窗口中计算结果
        PageViewBean bean = input.iterator().next();
        long pvCount = bean.getPvCount();
        long pvDuringTime = bean.getPvDuringTime();
        long uvCount = bean.getUvCount();
        long sessionCount = bean.getSessionCount();

        // 3输出
        String output = windowStartTime + "," + windowEndTime + "," + s
                + "," + pvCount + "," + pvDuringTime + "," + uvCount + "," + sessionCount
                + "," + System.currentTimeMillis();
        out.collect(output);


    }
}
